package com.suxin.module;

import com.suxin.module.bean.MyApi;
import com.suxin.module.bean.MyPoint;
import com.suxin.module.common.Json;
import com.suxin.module.http.MyHttpClient;
import com.suxin.module.http.MyHttpClientFactory;
import com.suxin.module.http.callback.MyBatchPutHttpResponseCallback;
import com.suxin.module.sender.consumer.MyConsumer;
import com.suxin.module.sender.consumer.MyConsumerImpl;
import com.suxin.module.sender.producer.MyProducer;
import com.suxin.module.sender.producer.MyProducerImpl;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @Classname MyOpenTSDBClient
 * @Description [ TODO ]
 * @Author Tang
 * @Date 2020/10/15 15:00
 * @Created by ASUS
 */
public class MyOpenTSDBClient {

    public final MyOpenTSDBConfig myOpenTSDBConfig;

    private MyHttpClient myHttpClient;

    private MyProducer myProducer;

    private MyConsumer myConsumer;

    private BlockingQueue<MyPoint> blockingQueue;

    public MyOpenTSDBClient(MyOpenTSDBConfig myOpenTSDBConfig) {
        this.myOpenTSDBConfig = myOpenTSDBConfig;
        try {
            this.myHttpClient = MyHttpClientFactory.clientHttpClient(myOpenTSDBConfig);
            this.myHttpClient.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
        // 不是只读的,创建生产者消费者模型
        if (!myOpenTSDBConfig.isReadonly()) {
            this.blockingQueue = new ArrayBlockingQueue<>(
                    myOpenTSDBConfig.getPutConsumerThreadCount()
            );
            this.myProducer = new MyProducerImpl(this.blockingQueue);
            this.myConsumer = new MyConsumerImpl(this.blockingQueue,
                    this.myHttpClient,
                    this.myOpenTSDBConfig);
            // 启动消费者
            this.myConsumer.start();
        }

    }

    /**
     * 写入数据
     * @param myPoint
     */
    public void put(MyPoint myPoint) {
        if (myOpenTSDBConfig.isReadonly()) {
            throw new IllegalArgumentException("this client is readonly,can't put point");
        }
        this.myProducer.send(myPoint);
    }

    /**
     * 同步写入，使用自定义callback
     * @param myPointList
     * @param batchPutCallBack
     */
    public void putSyncWithCallBack(List<MyPoint> myPointList,
                                    MyBatchPutHttpResponseCallback.BatchPutCallBack batchPutCallBack) {
        this.myHttpClient.post(
                MyApi.PUT.getPath(),
                Json.writeValueAsString(myPointList),
                new MyBatchPutHttpResponseCallback(batchPutCallBack,myPointList)
        );
    }



}